package poolserver

import (
	"bytes"
	"encoding/binary"
	"errors"
	"fmt"
	"net"
	"strconv"
	"time"

	"go.uber.org/zap"
)

func connectBackend(portal string, beUser string, bePasswd string, beDbName string) (net.Conn, error, uint32, uint32) {
	var conn net.Conn
	var err error
	var n int32
	var buf []byte
	var recvBuf []byte
	var pos int32

	var pid uint32 = 0
	var bekey uint32 = 0
	//var leftLen int32

	buf = make([]byte, 2048)
	recvBuf = make([]byte, 2048)
	//var portal = poolConf.GetNextPortal()
	d := net.Dialer{Timeout: 10 * time.Second}
	conn, err = d.Dial("tcp", portal)
	if err != nil {
		zap.S().Infof("Could not connect to %s: %s", portal, err.Error())
		return nil, err, 0, 0
	}

	//pos = 4
	//binary.BigEndian.PutUint16(recvBuf[pos:], 3)
	//pos += 2
	//binary.BigEndian.PutUint32(recvBuf[5:], 0)
	//pos += 2

	pos = 8

	connOpts := map[string]string{
		"user":                beUser,
		"database":            beDbName,
		"client_encoding":     "UTF8",
		"application_name":    "zqpool",
		"password_encryption": "md5",
	}

	for key, value := range connOpts {
		pos += appendString(buf[pos:], key)
		pos += appendString(buf[pos:], value)
	}
	buf[pos] = 0
	pos++
	//填包长度
	binary.BigEndian.PutUint32(buf[0:], uint32(pos))
	//填版本号，为3.0版本
	binary.BigEndian.PutUint16(buf[4:], 3)
	binary.BigEndian.PutUint16(buf[6:], 0)

	_, err = sendData(conn, buf[0:pos])
	if err != nil {
		zap.S().Infof("Send to db error: %s", err.Error())
		conn.Close()
		return nil, err, 0, 0
	}

	n, _, err = recvMessage(conn, recvBuf[0:])
	if err != nil {
		zap.S().Infof("Recv from db error: %s", err.Error())
		conn.Close()
		return nil, err, 0, 0
	}

	if recvBuf[0] != 'R' {
		zap.S().Infof("Expect from db recv 'R' message, but recv %c message: %v", buf[0], string(buf))
		conn.Close()
		return nil, errors.New("receive invalid packet"), 0, 0
	}
	authType := binary.BigEndian.Uint32(recvBuf[5:])
	if authType == 5 { /*type = 5 为md5验证*/

		/* concat('md5', md5(concat(md5(concat(password, username)), random-salt))) */

		md5AuthCalc(buf[5:], beUser, bePasswd, recvBuf[9:13])
		buf[0] = 'p'
		buf[40] = 0
		binary.BigEndian.PutUint32(buf[1:], 40)
		n, err = sendData(conn, buf[:41])
		if err != nil {
			zap.S().Infof("Send to PoolConfig error: %s", err.Error())
			conn.Close()
			return nil, err, 0, 0
		}
	} else if authType == 0 { /*type = 0 为trust验证, 代表连接成功不需要做处理*/

	} else {
		zap.S().Infof("Only support md5(type=5) and trust(type=0), can not support type = %d", authType)
		conn.Close()
		return nil, errors.New("only support md5(type=5)"), 0, 0
	}

	for {
		n, _, err = recvMessage(conn, buf[:])
		if err != nil {
			zap.S().Infof("Recv from Db error: %s", err.Error())
			conn.Close()
			return nil, err, 0, 0
		}
		if buf[0] == 'S' {
			//key, value := parseKeyValuePacket(recvBuf[1:])
			parseKeyValuePacket(buf[1:])
			//zap.S().Infof("Recv from PoolConfig parameter: %s=%s", key, value)
		} else if buf[0] == 'K' { /*BackendKeyData*/
			pid = binary.BigEndian.Uint32(buf[5:])
			bekey = binary.BigEndian.Uint32(buf[9:])
		} else if buf[0] == 'E' {
			n = int32(binary.BigEndian.Uint32(buf[1:]))
			zap.S().Infof("Recv from PoolConfig Error: %s", string(buf[5:n-1]))
			break
		} else if buf[0] == 'Z' { /* ReadyForQuery */
			//zap.S().Infof("Recv from PoolConfig ReadyForQuery, trans status is %c", recvBuf[5])
			break
		} else {
			zap.S().Debugf("Unknown message type: %c, msg length: %d", recvBuf[0], n)
		}
	}
	return conn, nil, pid, bekey
}

func sendPqReadyForQuery(conn net.Conn, transState byte) error {
	var buf [6]byte
	buf[0] = 'Z'
	binary.BigEndian.PutUint32(buf[1:], 5)
	buf[5] = transState
	_, err := sendData(conn, buf[:6])
	return err
}

func sendPqErrorResponse(conn net.Conn, errFields []string) error {
	var buf = make([]byte, 5, 128)
	buf[0] = 'E'
	binary.BigEndian.PutUint32(buf[1:], 0)
	for _, ef := range errFields {
		buf = append(buf, []byte(ef)...)
		buf = append(buf, 0)
	}
	buf = append(buf, 0)
	binary.BigEndian.PutUint32(buf[1:], uint32(len(buf)-1))
	_, err := sendData(conn, buf)
	return err
}

/*
func sendPqParseCompletion(conn net.Conn) error {
	var recvBuf [5]byte
	recvBuf[0] = '1'
	binary.BigEndian.PutUint32(recvBuf[1:], 4)
	_, err := sendData(conn, recvBuf[:5])
	return err
}*/

func sendPqCloseCompletion(conn net.Conn) error {
	var buf [5]byte
	buf[0] = '3'
	binary.BigEndian.PutUint32(buf[1:], 4)
	_, err := sendData(conn, buf[:5])
	return err
}

func sendPqClosePrepare(conn net.Conn, prepareId int32) error {
	var buf = make([]byte, 0, 12)
	buf = append(buf, 'C', 0, 0, 0, 0, 'S')
	buf = append(buf, []byte(fmt.Sprintf("S%d", prepareId))...)
	buf = append(buf, 0)
	binary.BigEndian.PutUint32(buf[1:], uint32(len(buf)-1))
	_, err := sendData(conn, buf)
	return err
}

func sendPqSync(conn net.Conn) error {
	var buf [5]byte
	buf[0] = 'S'
	binary.BigEndian.PutUint32(buf[1:], 4)
	_, err := sendData(conn, buf[:])
	return err
}

func sendRollback(conn net.Conn) error {
	var buf [15]byte
	buf[0] = 'Q'
	binary.BigEndian.PutUint32(buf[1:], 14)
	copy(buf[5:], "rollback;")
	buf[14] = 0
	_, err := sendData(conn, buf[:])
	return err
}

func printBackendErrorMessage(buf []byte) {
	var pos int32
	var begin int32
	var dataLen int32
	dataLen = int32(binary.BigEndian.Uint32(buf[1:5]))
	begin = 5
	pos = begin
	for pos < dataLen+1 {
		for ; pos < dataLen+1 && buf[pos] != 0; pos++ {
		}
		if pos-begin < 2 {
			return
		}
		//TCHDEBUG zap.S().Infof("%c(%s)", buf[begin], string(buf[begin+1:pos]))
		pos++
		begin = pos
	}
}

/*获得数据库是主库还是只读备库*/
func getDbReadWriteState(conn net.Conn) (error, int8) {
	var n int32
	var err error
	var isSuccess bool
	var packetLen int32
	var reqBuf []byte
	var recvBuf []byte
	sql := "select pg_is_in_recovery();"
	//var dataLen int32
	//var server_version string
	var errMsg string
	var pos int32
	var rw_state int8 = 0

	recvBuf = make([]byte, 1024)
	packetLen = int32(len(sql)) + 1 + 4
	reqBuf = make([]byte, packetLen+1+5)
	reqBuf[0] = 'Q'
	binary.BigEndian.PutUint32(reqBuf[1:], uint32(packetLen))
	pos = 5
	copy(reqBuf[pos:], []byte(sql))
	pos += int32(len(sql))
	reqBuf[pos] = '\000'
	pos++
	_, err = sendData(conn, reqBuf[:pos])
	if err != nil {
		return err, rw_state
	}

	isSuccess = true
	for {
		n, _, err = recvMessage(conn, recvBuf)
		if err != nil {
			return err, rw_state
		}

		if recvBuf[0] == 'E' {
			isSuccess = false
			printBackendErrorMessage(recvBuf[:n])
			errMsg = string(recvBuf)
		}

		//TCHDEBUG zap.S().Infof("Client(%d, B): RB(%d, %c): data=%v", ctx.Pid, ctx.pBackConn.Id, ctx.recvBuf[0], ctx.recvBuf[:n])
		if recvBuf[0] == 'Z' {
			break
		}

		if recvBuf[0] == 'D' { /*这是返回的一行数据*/
			//dataLen = int32(binary.BigEndian.Uint32(recvBuf[7:]))
			//server_version = string(recvBuf[11 : 11+dataLen])
			//第一个字节为D，
			//   4个字节为消息内容的长度（包括自身），
			//   2字节的列值的个数（可能是零）
			//   然后，为每个列都会出现下面的域对：
			//		Int32 列值的长度，以字节计（这个长度不包括它自己）。可以为零。一个特殊的情况是，-1表示一个NULL的域值。 如果是NULL的情况则后面不会跟着值字节。
			//         Byten 一个列的数值，以相关的格式代码指示的格式展现。n是上文的长度。

			if recvBuf[11] == 'f' {
				rw_state = 2
			} else {
				rw_state = 1
			}
		}
	}

	if isSuccess {
		return nil, rw_state
	} else {
		return errors.New(errMsg), rw_state
	}
}

/*判断SQL语句需要在主库上执行还是需要在备库上执行*/
func getStatementRwState(statement []byte) int8 {
	var rw_state int8 = 2
	if bytes.Equal(bytes.ToLower(statement[:12]), []byte("/*readonly*/")) {
		rw_state = 1
	}
	return rw_state
}

/*获得数据库服务器的版本*/
func getDbServerVersion(conn net.Conn) (error, string) {
	var n int32
	var err error
	var isSuccess bool
	var packetLen int32
	var reqBuf []byte
	var recvBuf []byte
	sql := "show server_version;"
	var dataLen int32
	var server_version string
	var errMsg string
	var pos int32

	recvBuf = make([]byte, 1024)
	packetLen = int32(len(sql)) + 1 + 4
	reqBuf = make([]byte, packetLen+1+5)
	reqBuf[0] = 'Q'
	binary.BigEndian.PutUint32(reqBuf[1:], uint32(packetLen))
	pos = 5
	copy(reqBuf[pos:], []byte(sql))
	pos += int32(len(sql))
	reqBuf[pos] = '\000'
	pos++
	_, err = sendData(conn, reqBuf[:pos])
	if err != nil {
		return err, ""
	}

	isSuccess = true
	for {
		n, _, err = recvMessage(conn, recvBuf)
		if err != nil {
			return err, ""
		}

		if recvBuf[0] == 'E' {
			isSuccess = false
			printBackendErrorMessage(recvBuf[:n])
			errMsg = string(recvBuf)
		}

		//TCHDEBUG zap.S().Infof("Client(%d, B): RB(%d, %c): data=%v", ctx.Pid, ctx.pBackConn.Id, ctx.recvBuf[0], ctx.recvBuf[:n])
		if recvBuf[0] == 'Z' {
			break
		}

		if recvBuf[0] == 'D' { /*这是返回的一行数据*/
			dataLen = int32(binary.BigEndian.Uint32(recvBuf[7:]))
			server_version = string(recvBuf[11 : 11+dataLen])
		}
	}

	if isSuccess {
		return nil, server_version
	} else {
		return errors.New(errMsg), ""
	}
}

/*获得数据库是主库还是只读备库*/
func getTimelineAndLsn(conn net.Conn) (error, int32, int64) {
	var n int32
	var err error
	var isSuccess bool
	var packetLen int32
	var reqBuf []byte
	var recvBuf []byte
	sql := "select timeline_id, pg_wal_lsn_diff(pg_current_wal_lsn(), '0/0'::pg_lsn)::bigint from pg_control_checkpoint();"
	var dataLen int32
	//var server_version string
	var errMsg string
	var pos int32
	var timeline_id int32 = 0
	var lsn int64 = 0
	var i64 int64

	recvBuf = make([]byte, 1024)
	packetLen = int32(len(sql)) + 1 + 4
	reqBuf = make([]byte, packetLen+1+5)
	reqBuf[0] = 'Q'
	binary.BigEndian.PutUint32(reqBuf[1:], uint32(packetLen))
	pos = 5
	copy(reqBuf[pos:], []byte(sql))
	pos += int32(len(sql))
	reqBuf[pos] = '\000'
	pos++
	_, err = sendData(conn, reqBuf[:pos])
	if err != nil {
		return err, timeline_id, 0
	}

	isSuccess = true
	for {
		n, _, err = recvMessage(conn, recvBuf)
		if err != nil {
			return err, timeline_id, 0
		}

		if recvBuf[0] == 'E' {
			isSuccess = false
			printBackendErrorMessage(recvBuf[:n])
			errMsg = string(recvBuf)
		}

		//TCHDEBUG zap.S().Infof("Client(%d, B): RB(%d, %c): data=%v", ctx.Pid, ctx.pBackConn.Id, ctx.recvBuf[0], ctx.recvBuf[:n])
		if recvBuf[0] == 'Z' {
			break
		}

		if recvBuf[0] == 'D' { /*这是返回的一行数据*/
			//dataLen = int32(binary.BigEndian.Uint32(recvBuf[7:]))
			//server_version = string(recvBuf[11 : 11+dataLen])
			//第一个字节为D，
			//   4个字节为消息内容的长度（包括自身），
			//   2字节的列值的个数（可能是零）
			//   然后，为每个列都会出现下面的域对：
			//		Int32 列值的长度，以字节计（这个长度不包括它自己）。可以为零。一个特殊的情况是，-1表示一个NULL的域值。 如果是NULL的情况则后面不会跟着值字节。
			//         Byten 一个列的数值，以相关的格式代码指示的格式展现。n是上文的长度。
			pos = 7
			dataLen = int32(binary.BigEndian.Uint32(recvBuf[pos:]))
			pos += 4
			if dataLen == -1 { // 这是NULL值
				timeline_id = 0
			} else {
				i64, err = strconv.ParseInt(string(recvBuf[pos:pos+dataLen]), 10, 32)
				if err != nil {
					isSuccess = false
					errMsg = err.Error()
				}
				pos += dataLen
				timeline_id = int32(i64)
			}
			dataLen = int32(binary.BigEndian.Uint32(recvBuf[pos:]))
			pos += 4
			if dataLen == -1 { // 这是NULL值
				lsn = 0
			} else {
				lsn, err = strconv.ParseInt(string(recvBuf[pos:pos+dataLen]), 10, 64)
				if err != nil {
					isSuccess = false
					errMsg = err.Error()
				}
			}
		}
	}

	if isSuccess {
		return nil, timeline_id, lsn
	} else {
		return errors.New(errMsg), timeline_id, lsn
	}
}

/*获得数据库服务器的版本*/
func setApplicationName(conn net.Conn, AppName string) error {
	var n int32
	var err error
	var isSuccess bool
	var packetLen int32
	var reqBuf []byte
	var recvBuf []byte
	var sql string
	//var dataLen int32
	//var server_version string
	var errMsg string
	var pos int32

	sql = "set application_name to '" + AppName + "';"
	recvBuf = make([]byte, 1024)
	packetLen = int32(len(sql)) + 1 + 4
	reqBuf = make([]byte, packetLen+1+5)
	reqBuf[0] = 'Q'
	binary.BigEndian.PutUint32(reqBuf[1:], uint32(packetLen))
	pos = 5
	copy(reqBuf[pos:], []byte(sql))
	pos += int32(len(sql))
	reqBuf[pos] = '\000'
	pos++
	_, err = sendData(conn, reqBuf[:pos])
	if err != nil {
		return err
	}

	isSuccess = true
	for {
		n, _, err = recvMessage(conn, recvBuf)
		if err != nil {
			return err
		}

		if recvBuf[0] == 'E' {
			isSuccess = false
			printBackendErrorMessage(recvBuf[:n])
			errMsg = string(recvBuf)
		}

		//TCHDEBUG zap.S().Infof("Client(%d, B): RB(%d, %c): data=%v", ctx.Pid, ctx.pBackConn.Id, ctx.recvBuf[0], ctx.recvBuf[:n])
		if recvBuf[0] == 'Z' {
			break
		}
	}

	if isSuccess {
		return nil
	} else {
		return errors.New(errMsg)
	}
}

// SendCancelRequest 向数据库服务器发送CancelRequest
func SendCancelRequest(portal string, pid uint32, bekey uint32) error {
	var err error
	var reqBuf []byte
	var conn net.Conn

	d := net.Dialer{Timeout: 10 * time.Second}
	conn, err = d.Dial("tcp", portal)
	if err != nil {
		return err
	}

	reqBuf = make([]byte, 16)
	binary.BigEndian.PutUint32(reqBuf[0:], 16)
	binary.BigEndian.PutUint16(reqBuf[4:], 1234)
	binary.BigEndian.PutUint16(reqBuf[6:], 5678)
	binary.BigEndian.PutUint32(reqBuf[8:], pid)
	binary.BigEndian.PutUint32(reqBuf[12:], bekey)
	_, err = sendData(conn, reqBuf[:16])
	conn.Close()
	return err
}

// CleanupTrans 清理事务
func CleanupTrans(conn net.Conn) error {
	var err error
	err = sendRollback(conn)
	if err != nil {
		return err
	}

	var buf [1024]byte
	for {
		_, _, err = recvMessage(conn, buf[:])
		if err != nil {
			return err
		}
		if buf[0] == 'Z' { /*ReadyForQuery*/
			return nil
		}
	}
	//return nil
}
